%% Author: user 
%% Created: 2006-8-22
%% Description: TODO: Add desciption to aleph_dht_rpc
-module(aleph_dht_rpc).
-include("aleph_data_interface.hrl").
-include("aleph_service_macros.hrl").
-include("aleph_naming_macros.hrl").
-behaviour(gen_server).
%% Exported Functions
-export([init/1,start_link/2,start/2,
         handle_call/3,handle_cast/2,handle_info/2,terminate/2,
         send/3,process_spec/1,close/0,protocol/3,protocol/4]).
-import(aleph_config,[get_reply_time_out/0]).
-import(aleph_process_labeler,[label_process/1]).
-import(aleph_utility,[is_expired/1]).
  
%%%%%%%%%%%%%public functions
process_spec(Args)->
      {?DHT_RPC_SERVER,{aleph_dht_rpc,start_link,Args},transient, brutal_kill,worker, [aleph_dht_rpc]}.

protocol(Type,Method,Args)->
    protocol:create(Type,Method,[Args],"","").

protocol(Type,Method,Args,SessionKey)->
    protocol:create(Type,Method,[Args],"",SessionKey).
send(Ip,Port,Msg)when ?IS_OBJECT(Msg)==true  ->
    gen_server:call(?DHT_RPC_SERVER,{out_going,Ip,Port,Msg});
send(Ip,Port,Msg)->
    gen_server:call(?DHT_RPC_SERVER,{out_going,Ip,Port,protocol:object(Msg)}).

close()->
    gen_server:cast(?DHT_RPC_SERVER,close).
%%%%%%%%%%%%%behaviour implementations
start_link(Port,HModule) ->    
    label_process(
                  gen_server:start_link({local,?DHT_RPC_SERVER}, aleph_dht_rpc, [Port,HModule], [])).
    
    
start(Port,HModule)->
    label_process(
                  gen_server:start({local,?DHT_RPC_SERVER},aleph_dht_rpc,[Port,HModule],[] )).
%%%%%%%%%%%%%callback implementations
init([Port,HModule]) -> 
    process_flag(trap_exit, true),
    case gen_udp:open(Port,[binary]) of
       {ok,Socket}->{ok,{Socket,HModule}};
       {error,Reason}->{stop,Reason}
    end.
 
handle_cast(close,State)->
    erlang:exit(self(),normal),
    {noreply, State}.

handle_call({out_going,Ip,Port,Msg}, _From,State={Socket,_})->
   case timer:send_after(get_reply_time_out(),self(),{time_out,Ip,Port,Msg}) of
        {ok,Timer}->              
           gen_udp:send(Socket,Ip,Port,change_timer(Msg,Timer)),
           {reply,{ok,Timer},State};               
        {error,Reason}->
           {reply,{error,Reason},State} 
    end.   

handle_info({udp,_,Ip,Port,Packet},State={Socket,HModule}) ->
    %io:format("got udp ~p~n",[Packet]),
    do_packet(binary_to_term(Packet),Ip,Port,Socket,HModule),
    {noreply, State};

handle_info({time_out,_,_,_},State) ->
    io:format("time out ~n"), 
    {noreply, State};

handle_info({'EXIT',_,Reason}, State) ->
    {stop, Reason, State}. 

terminate(_,{Socket,_})->
    gen_udp:close(Socket),
    io:format("socket closed"),
    ok.

%%%%%%%%%%%%%private function   
change_timer(Msg,Timer)->    
     term_to_binary((Msg({timer,Timer}))(self)).

error_msg(Reason,Timer,Fun)->
     term_to_binary(protocol:new(err,Fun,[Reason],Timer,1)).

do_packet(Packet,Ip,Port,S,HModule)->    
    #protocol{typ=Type,args=Args,method=Fun,timer=Timer_ref,session_key=Key}=Packet,
    %io:format("do Packet ~p~n",[Packet]),   
    try
      case Type of
              rsp->    
                  case is_expired(Timer_ref) of                      
                  true-> 
                      timer:cancel(Timer_ref),    
                      HModule:do_actions(Fun,[{Ip,Port}|Args]);                      
                  false->
                      gen_udp:send(S,Ip,Port,error_msg("message timer is expired",Timer_ref,Fun))
                  end;                     
              req->
                  %io:format("Call Fun ~p Args ~p~n",[Fun,[{Ip,Port}|Args]]),
                  Ret=if is_binary(Key)==true->
                         HModule:do_actions(Fun,[{Ip,Port},Key|Args]);
                       true->
                         HModule:do_actions(Fun,[{Ip,Port}|Args])
                      end,                          
                  %io:format("End Call~n"),
                  NewRet=Ret({session_key,Key}),
                  Response=change_timer(NewRet,Timer_ref),
                  gen_udp:send(S,Ip,Port,Response)
      end
    catch
       error:Reason->io:format("error ~p~n",[Reason]),
       gen_udp:send(S,Ip,Port,error_msg(Reason,Timer_ref,Fun))
    end.        
 
